package com.etl.component.basic;

import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 描述：
 *
 * @author xianggj
 * @Date 2021/12/6 17:30
 **/
public class InterFunction implements ReduceFunction<Tuple2<Integer,Row>> {

    private Integer[] indexs;

    TypeInformation<?>[] fieldTypes;

    private Row beforRow;

    /**
     * 缓存修改值
     */
    private List<Tuple2<Integer,Row>> cacheRows = new ArrayList<>();

    /**
     * 结果值
     */
    private List<Row> resList;

    public List<Row> getResList() {
        if (cacheRows.size() > 0){
            addResCache();
        }
        return resList;
    }

    /**
     * 添加缓存到resList
     */
    private void putCache(int i,  Object field) {
        if (cacheRows.size() == 0 || field == null){
            return;
        }
        //只有当前列为空值才是要替换的值
        List<Tuple2<Integer, Row>> collect =
                cacheRows.stream().filter(e -> e.f1.getField(i) == null).collect(Collectors.toList());
        if (collect.size() == 0){
            return;
        }
        //计算跨度
        Object resVal = null;
        if (beforRow == null){
            resVal = field;
        } else {
            Object beforRowVal = beforRow.getField(i);
            if (beforRowVal == null){
                resVal = field;
            } else {
                Double eDouble = Double.valueOf(beforRowVal.toString());
                Double bDouble = Double.valueOf(field.toString());
                resVal = (bDouble - eDouble)/(collect.size()+1);
            }
        }
        //填充跨度
        for (int j = 0; j < collect.size(); j++) {
            Tuple2<Integer, Row> tuple2 = collect.get(j);
            Row e = tuple2.f1;
            Object oldVal = e.getField(i);
            if (oldVal != null){
                continue;
            }
            if (beforRow == null){
                e.setField(i, resVal);
            } else {
                Object beforRowVal = beforRow.getField(i);
                if (beforRowVal == null){
                    e.setField(i, resVal);
                } else {
                    Double eDouble = Double.valueOf(beforRowVal.toString());
                    TypeInformation<?> fieldType =  fieldTypes[i];
                    Double aDouble = Double.valueOf(resVal.toString());
                    e.setField(i, AbstractFlinkComponent.covert(eDouble + ((j+1)*aDouble), fieldType));
                }
            }
        }
        //校验所有数据是否填充完成
        for (int j = 0; j < cacheRows.size() ; j++) {
            Tuple2<Integer, Row> tuple2 = cacheRows.get(j);
            Row e = tuple2.f1;
            boolean isComplate = true;
            for (Integer index : indexs) {
                Object val = e.getField(index);
                if (val == null) {
                    isComplate = false;
                }
            }
            if (isComplate){
                resList.add(tuple2.f0,e);
                cacheRows.remove(j);
                j--;
            }

        }

    }



    /**
     * 最后添加缓存到resList
     */
    private void addResCache() {
        for (Integer in : indexs) {
            Object field = beforRow.getField(in);
            if (field != null) {
                cacheRows.stream().forEach(e->{
                    e.f1.setField(in, field);
                });
            }
        }
        cacheRows.stream().forEach(e->{
            resList.add(e.f0, e.f1);
        });
    }

    public InterFunction(Integer[] indexs, TypeInformation<?>[] fieldTypes, int size) {
        this.indexs = indexs;
        this.fieldTypes = fieldTypes;
        this.resList = new ArrayList<>(size);
    }

    @Override
    public Tuple2<Integer,Row> reduce(Tuple2<Integer,Row> row, Tuple2<Integer,Row> t1) throws Exception {
        boolean isNeedCacha = false;
        //第一个特殊处理
        if (resList.size() == 0 && cacheRows.size() == 0){
            for (Integer in : indexs) {
                Object field = row.f1.getField(in);
                if (field == null) {
                    isNeedCacha = true;
                }
            }
            if (isNeedCacha){
                cacheRows.add(row);
            } else {
                resList.add(row.f0 ,row.f1);
            }
            beforRow = Row.join(row.f1);
        }
        for (Integer in : indexs) {
            Object field = t1.f1.getField(in);
            if (field == null) {
                isNeedCacha = true;
            } else {
                //添加值
                putCache(in, field);
                //更新值
                beforRow.setField(in, field);
            }
        }
        if (isNeedCacha){
            cacheRows.add(t1);
        } else {
            resList.add(t1.f0, t1.f1);
        }
        return t1;
    }
}
